package com.yanfan.iot.ruleEngine;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.yanfan.common.utils.spring.SpringUtils;
import com.yanfan.framework.manager.AsyncManager;
import com.yanfan.framework.utils.RedisUtils;
import com.yanfan.iot.domain.MultipleDataSource;
import com.yanfan.mqttclient.PubMqttClient;
import com.yomahub.liteflow.exception.NullParamException;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.text.StringSubstitutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.SQLException;
import java.sql.Statement;
import java.time.Duration;
import java.util.Objects;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;


@Data
@Builder
@Slf4j
@NoArgsConstructor
@AllArgsConstructor
public class MsgContext {

    // 在规则引擎中使用，这个不能删
    private static Logger logger = LoggerFactory.getLogger("script");

    /**
     * 消息主题
     */
    private String topic;

    /**
     * 默认主题
     */
    private String defaultTopic;

    /**
     * 消息内容
     */
    private String payload;
    /**
     * 设备编号
     */
    private String serialNumber;
    /**
     * 产品id
     */
    private Long productId;
    /**
     * 协议编码
     */
    private String protocolCode;

    private ConcurrentHashMap<String, Object> dataMap = new ConcurrentHashMap<>();

    private <T> void putDataMap(String key, T t) {
        if (this.dataMap == null) {
            this.dataMap = new ConcurrentHashMap<>();
        }
        if (ObjectUtil.isNull(t)) {
            throw new NullParamException("data can't accept null param");
        } else {
            this.dataMap.put(key, t);
        }
    }

    public boolean hasData(String key) {
        if (this.dataMap == null) {
            return false;
        }
        return this.dataMap.containsKey(key);
    }

    public <T> T getData(String key) {
        if (this.dataMap == null) {
            return null;
        }
        return (T) this.dataMap.get(key);
    }

    public <T> void setData(String key, T t) {
        this.putDataMap(key, t);
    }

    //"{ topic:${topic}, payload:${payload} }";
    // 自定义占位符，可在脚本中使用msgContext.setData("test":1);
    // 然后通过${test}调用，可在输出侧http body和 sql语句中传入该值
    public String placeholders(String str) {
        if (getData("topic") == null) {
            setData("topic", getTopic());
        }
        if (getData("payload") == null) {
            setData("payload", getPayload());
        }
        if (getData("serialNumber") == null) {
            setData("serialNumber", getSerialNumber());
        }
        if (getData("productId") == null) {
            setData("productId", getProductId());
        }
        if (getData("protocolCode") == null) {
            setData("protocolCode", getProtocolCode());
        }
        StringSubstitutor substitutor = new StringSubstitutor(this.dataMap);
        return substitutor.replace(str);
    }

    public void databaseStorage(Long id) {
        DataSourceService dataSourceService = SpringUtils.getBean(DataSourceService.class);
        MultipleDataSource multipleDataSource = dataSourceService.getDataSource(id);
        DruidDataSource dataSource = dataSourceService.createDataSource(multipleDataSource);
        DruidPooledConnection connection = null;
        Statement statement = null;
        String sql = placeholders(multipleDataSource.getSql());
        log.info("执行sql：{}", sql);
        try {
            connection = dataSource.getConnection();
            statement = connection.createStatement();
            statement.execute(sql);
        } catch (SQLException e) {
            log.error("sql执行失败---------，原因：{ }", e);
        }
        try {
            connection.close();
            if (statement != null) {
                statement.close();
            }
        } catch (SQLException e) {
            log.error("关闭数据库连接失败，原因：{ }", e);
        }
    }


    /**
     * 规则引擎消息分发，在规则引擎脚本中调用
     * 示例：
     * msgContext.sendMessage("/1114/D104220T7S7D/property/post", "[{
     * "id": "power",
     * "value": "44"
     * }]");
     */
    private void sendMessage(String topic, String payload) {
        String cacheKey = "msgContext-sendMessage:" + topic;
        Boolean isSend = RedisUtils.getCacheObject(cacheKey);
        // 重复的消息不重复执行，1秒之内只允许发送一次，防止桥接的mqtt topic 与 自己的服务器主题一致，导致出现循环
        if (Objects.isNull(isSend) || !isSend) {
            log.info("规则引擎分发消息：{}", topic);
            AsyncManager.me().execute(new TimerTask() {
                @Override
                public void run() {
                    PubMqttClient mqttClient = SpringUtils.getBean(PubMqttClient.class);
                    mqttClient.publish(0, false, topic, payload);
                    RedisUtils.setCacheObject(cacheKey, true, Duration.ofSeconds(1));
                }
            });
        } else {
            log.info("规则引擎分发消息重复：{}, payload: {}", topic, payload);
        }

    }
}
